package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MykafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        // TODO: 1. 基本环境准备
        // 1.1 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 1.2 设置并行度
        env.setParallelism(4);
        // TODO: 2.设置检查点
/*        // 2.1 每隔5s创建一次检查点
        env.enableCheckpointing(5 * 1000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置检查点超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        // 2.3 设置取消job后是否保存检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.4 设置检查点重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));

        // 2.5 设置状态后端 内存|文件系统|RockDB
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/Fsck", false));
        env.setStateBackend(new RocksDBStateBackend("hdfs://hadoop102:8020/Flink/gmall/CK2", false));
//         2.6 设置hadoop用户
        System.setProperty("HADOOP_USER_NAME", "atguigu");
//        TODO 3.从Kafka中读取数据*/
//        3.1 声明消费的主题以及消费者组
        String topic = "ods_base_log";
        String groupId = "base_log_app_group";
//        3.2 通过工具类 获取Kafka消费者对象
        FlinkKafkaConsumer<String> KafkaSource = MykafkaUtil.getkafkaSource(topic, groupId);
//        3.3 将消费到的数据封装到流中
        DataStreamSource<String> KafkaDS = env.addSource(KafkaSource);
//        3.4 对流中的数据结构进行转换
//        3.4.1 匿名内部类
        KafkaDS.map(

                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String value) throws Exception {
                        return JSON.parseObject(value);
                    }
                }
        );
//        3.4.2 lambda表达式
        SingleOutputStreamOperator<JSONObject> jsonOjDS = KafkaDS.map(jsonStr -> JSON.parseObject(jsonStr));
//        3.4.3 方法的默认调用
//        SingleOutputStreamOperator<JSONObject> jsonobjDs = KafkaDS.map(JSON::parseObject);
        // TODO: 4. 新老顾客状态恢复
        /**
         * 思路:将设备上次访问时间记录到状态中，下次再访问的时候，从状态中获取时间,如果已经访问过,修复；
         * 如果没有,说明没有访问过,将这次访问的时间记录到状态中
         *
         *
         */
//       4.1 按照mid进行分组
        KeyedStream<JSONObject, String> keyedDS = jsonOjDS.keyBy(jsonobj -> jsonobj.getJSONObject("common").getString("mid"));
//        4.2 状态修复
        SingleOutputStreamOperator<JSONObject> jsonObjWithFlagDs = keyedDS.map(

                new RichMapFunction<JSONObject, JSONObject>() {
                    //todo 注意 不能在声明的时候直接初始化,这个时候获取不到getTRuntimeContext
                    private ValueState<String> lastVisitDateState;
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        sdf = new SimpleDateFormat("yyyyMMdd");
                        lastVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("state", Types.STRING));
                    }

                    @Override
                    public JSONObject map(JSONObject value) throws Exception {
//                        获取状态标记
                        String isNew = value.getJSONObject("common").getString("is_new");
//                        获取当前访问日期
                        String curDate = sdf.format(value.getLong("ts"));
                        // 如果标记是新访客,那么状态有可能是不准确的,需要修复
                        if (isNew.equals("1")) {
                            // 从Flink状态中获取上次访问的日期
                            String lastVisitDate = lastVisitDateState.value();
                            if (lastVisitDate != null && lastVisitDate.length() > 0) {  // 该访客已经访问过
                                if (!lastVisitDate.equals(curDate)) { // 在今日之前已经访问过      >>>> 预设:只要在首次访问发生在今日内 都算新访客
                                    // 说明已经访问过,进行状态的修复
                                    isNew = "0";
                                    value.getJSONObject("common").put("is_new", isNew);

                                }
                            }
                        } else {
                            //todo 说明该设备是今天第一次访问,将当前访问日期放到状态中
                            lastVisitDateState.update(curDate);
                        }
                        return value;
                    }
                }
        );
        // TODO: TODO 5. 分流 启动--> 启动侧输出流 ；曝光 --> 曝光侧输出流 ；页面--> 主流
//        5.1 定义侧输出标记
        OutputTag<String> startTg = new OutputTag<String>("startTag") {
        };
        OutputTag<String> dispalyTag = new OutputTag<String>("dispalyTag") {
        };
//      5.2分流
        SingleOutputStreamOperator<String> splitDS = jsonObjWithFlagDs.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObject, Context ctx, Collector<String> out) throws Exception {
//                        获取启动属性
                        JSONObject startjsonObj = jsonObject.getJSONObject("start");
//                        将接收到的jsonObj转为字符串
                        String jsonStr = jsonObject.toJSONString();
//                        判断是否为启动日志
                        if (startjsonObj != null && startjsonObj.size() > 0) {
//                            是启动日志 将启动日志数据放入启动侧输出流中
                            ctx.output(startTg, jsonStr);

                        } else {//如果不是启动日志，那么就是页面日志
                            out.collect(jsonStr);
//                            同时在页面日志中还包含曝光日志
                            JSONArray displaysArr = jsonObject.getJSONArray("displays");
//                            判断是否为曝光日志
                            if (displaysArr != null && displaysArr.size() > 0) {
//                                获取页面ID
                                String pageId = jsonObject.getJSONObject("page").getString("page_id");
//                                对曝光数组进行遍历
                                for (int i = 0; i < displaysArr.size(); i++) {
                                    JSONObject displaysArrJSONObject = displaysArr.getJSONObject(i);
//                                    给曝光的json对象补充pageId
                                    displaysArrJSONObject.put("page_id", pageId);
                                    ctx.output(dispalyTag, displaysArrJSONObject.toJSONString());
                                }
                            }
                        }
                    }
                }
        );
//        5.3 获取各个流数据并输出测试
        DataStream<String> startDS = splitDS.getSideOutput(startTg);
        DataStream<String> displayDS = splitDS.getSideOutput(dispalyTag);
        splitDS.print("页面");
        startDS.print("###启动###");
        displayDS.print("$$$曝光$$$");
// TODO 6. 将不同的流的数据写道kafka的dwd的不同主题中
        // 6.1 声明kafka主题
        String startSinkTopic = "dwd_start_log";
        String displaySinkTopic = "dwd_display_log";
        String pageSinkTopic = "dwd_page_log";
        // 6.2 Sink 操作
        startDS.addSink(MykafkaUtil.getKafkaSink(startSinkTopic));
        displayDS.addSink(MykafkaUtil.getKafkaSink(displaySinkTopic));
        splitDS.addSink(MykafkaUtil.getKafkaSink(pageSinkTopic));
        env.execute();
    }
}
